#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include "cJSON.h"
#include "MQTTClient.h"
#include "unistd.h"
#define ADDRESS "tcp://localhost:1883"  // MQTT代理服务器地址
#define PUB_CLIENTID "ExampleClientPub" // 发布者客户端ID
#define SUB_CLIENTID "ExampleClientSub" // 订阅者客户端ID
#define PUB_TOPIC "/app/data/up"        // 发布主题
#define SUB_TOPIC "/app/data/down"      // 订阅主题
#define QOS         1
#define TIMEOUT     10000L

volatile MQTTClient_deliveryToken deliveredtoken;

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");

    cJSON *root = cJSON_Parse(message->payload);
    if(root)
    {
        char * json = cJSON_Print(root);
        if(json)
        {
            printf("%s\n",json);
            free (json);
        }
        cJSON_Delete(root);
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

void subscribe_to_mqtt()
{
    MQTTClient client;                                                           // 客户端句柄（描述符）
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer; // 初始化连接选项
    int rc;                                                                      // 返回值

    // 创建客户端，并且指定客户端连接的mqtt服务器地址和客户端ID
    MQTTClient_create(&client, ADDRESS, SUB_CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20; // 设置保持连接的间隔为20秒
    conn_opts.cleansession = 1;       // 设置为1表示每次连接都是一个干净的会话，不会保留之前的订阅信息

    // 设置回调函数，用于处理连接丢失、消息到达和消息送达的情况
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    // 尝试连接到MQTT代理，如果连接失败则打印错误信息并退出程序
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("连接失败，返回代码 %d\n", rc);
        exit(EXIT_FAILURE);
    }

    // 订阅指定的主题
    MQTTClient_subscribe(client, SUB_TOPIC, QOS);

    while (1)
    {
        MQTTClient_yield(); // 调用yield函数，处理等待的消息
        sleep(1);           // 休眠1秒，减少CPU使用
    }

    // MQTTClient_unsubscribe(client, SUB_TOPIC);
    // MQTTClient_disconnect(client, 10000);
    // MQTTClient_destroy(&client);
}
void *thread_subscribe(void *arg)
{
    subscribe_to_mqtt();
    return NULL;
}


void publish_to_mqtt(/*const char *message*/)
{
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;

    MQTTClient_create(&client, ADDRESS, PUB_CLIENTID,
        MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    while(1)
    {	    // 从用户读取输入
    char json_string[10000];
    printf("Enter JSON data to publish: ");
    // 读取 JSON 字符串
    if (fgets(json_string, sizeof(json_string), stdin) == NULL) {
        printf("Error reading input.\n");
        MQTTClient_disconnect(client, 10000);
        MQTTClient_destroy(&client);
	return;
    }

    // 去除输入尾部换行符
    json_string[strcspn(json_string, "\n")] = '\0'; 
    
    // 设置消息内容并发布
    pubmsg.payload = json_string;
    pubmsg.payloadlen = (int)strlen(json_string);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, PUB_TOPIC, &pubmsg, &token);
    printf("Waiting for up to %d seconds for publication of %s\n"
            "on topic %s for client with ClientID: %s\n",
            (int)(TIMEOUT/1000), json_string, PUB_TOPIC, PUB_CLIENTID);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("Message with delivery token %d delivered\n", token);
    }
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
}

void *thread_publish(void *arg)
{
    
    publish_to_mqtt(/*const char *message*/);
    return NULL;
}

int main()
{
    pthread_t tid_subscribe, tid_publish;

    pthread_create(&tid_subscribe,NULL,thread_subscribe, NULL);
    pthread_create(&tid_publish, NULL,thread_publish, NULL);

    pthread_join(tid_subscribe, NULL);
    pthread_join(tid_publish, NULL);

    return 0;
}
